package com.wyh.apitest.transformation.keyedflow

import com.wyh.apitest.source.SensorReading
import org.apache.flink.api.common.functions.ReduceFunction
import org.apache.flink.streaming.api.scala._

object keyedflowDemo {
  def main(args: Array[String]): Unit = {
    //定义流式处理环境
    val streamEnv = StreamExecutionEnvironment.getExecutionEnvironment
    streamEnv.setParallelism(1)

    val sensorinfo: DataStream[String] = streamEnv.readTextFile("D:\\IdeaProjects\\FlinkStudyScala\\src\\main\\resources\\sensor.txt")
    //转换成样例类类型
    val dataStream = sensorinfo.map(s => {
      val arr: Array[String] = s.split(",")
      SensorReading(arr(0), arr(1).toLong, arr(2).toDouble)
    })


    /**
      * keyby 和min
      * 如果是min最后只是那个字段数值是最小其他字段是读进来的值
      * 如果是minby，最后输出的是最小的对象全部字段
      * 分组聚合，输出每个传感器温度最小值
      */
//    val stream1 = dataStream.keyBy("id") //根据id分组
//      //      .min("temperature")
//      .minBy("temperature")
//    stream1.print("keyby操作：")

    /**
      * reduce
      * 计算到目前最新的时间为止，温度最小的那个对象值
      */
    val stream2 = dataStream.keyBy("id")
//      .reduce((curState, newData) => {
//        SensorReading(curState.id, newData.timestamp, curState.temperature.min(newData.temperature))
//      })
        .reduce(new MyReduceFunction)
    stream2.print("reduce操作：")


    streamEnv.execute("键控流转换操作")

  }

}

class MyReduceFunction extends ReduceFunction[SensorReading]{
  override def reduce(t: SensorReading, t1: SensorReading): SensorReading = {
    SensorReading(t.id,t1.timestamp,t.temperature.min(t1.temperature))
  }
}